﻿//Copyright (c) Microsoft Corporation.  All rights reserved.

using System;

namespace Microsoft.Ted.Wacel.Base
{
    public class Queue<T> : Array<T>
    {
        private long mSize;
        private string mHeadIndexKey;
        private string mTailIndexKey;
        
        public Queue(string name, IDataProvider dataProvider, ILockProvider lockProvider):
            this(name, long.MaxValue - 1, dataProvider, lockProvider)
        {
        }

        public Queue(string name,  long maxLength, IDataProvider dataProvider, ILockProvider lockProvider)
            : base(name, maxLength, dataProvider, lockProvider)
        {
            if (maxLength == long.MaxValue)
                throw new ArgumentOutOfRangeException("Maximum allowed queue size is " + (long.MaxValue - 1));
            mSize = maxLength;
            mHeadIndexKey = name + "_head";
            mTailIndexKey = name + "_tail";
        }

        public void Enqueue(T item)
        {
            object tailHandle;
            object headHandle;
            
            long tailIndex = ReadAndLock(mTailIndexKey, -1L, TimeSpan.FromSeconds(100), out tailHandle);
            long headIndex = ReadAndLock(mHeadIndexKey, -1L, TimeSpan.FromSeconds(100), out headHandle);

            try
            {
                var length = tailIndex - headIndex;
                if (length < 0)
                    length += mSize + 1;
                else if (length == mSize + 1)
                    length = mSize;
                if (length >= mSize)
                    throw new InvalidOperationException("Queue is full.");

                tailIndex = (tailIndex + 1) % (mSize + 1);

                if (headIndex >= 0 && tailIndex == headIndex)
                    throw new InvalidOperationException("Queue is full.");

                Put(tailIndex.ToString(), item);
            }
            catch
            {
                throw;
            }
            finally
            {
                try
                {
                    PutAndUnlock(mTailIndexKey, tailIndex, tailHandle);
                }
                catch (DataProviderException)
                {

                }
                try
                {
                    Unlock(mHeadIndexKey, headHandle);
                }
                catch (DataProviderException)
                {

                }
            }
        }

        public T Dequeue()
        {
            object tailHandle;
            object headHandle;

            long tailIndex = ReadAndLock(mTailIndexKey, -1L, TimeSpan.FromSeconds(100), out tailHandle);
            long headIndex = ReadAndLock(mHeadIndexKey, -1L, TimeSpan.FromSeconds(100), out headHandle);
            try
            {
                long peekHeadIndex = (headIndex + 1) % (mSize + 1);

                if (peekHeadIndex > tailIndex)
                    throw new InvalidOperationException("Queue is empty.");
                else
                    headIndex = peekHeadIndex;

                return (T)Get(headIndex.ToString(),typeHint:typeof(T));
            }
            catch
            {
                throw;
            }
            finally
            {
                try
                {
                    PutAndUnlock(mHeadIndexKey, headIndex, headHandle);
                }
                catch (DataProviderException)
                {

                }
                try
                {
                    Unlock(mTailIndexKey, tailHandle);
                }
                catch (DataProviderException)
                {

                }
            }
        }
        public new long Length
        {
            get
            {
                var length = InterlockedRead(mTailIndexKey, 0) - InterlockedRead(mHeadIndexKey, 0);
                if (length < 0)
                    length += mSize + 1;
                else if (length == mSize + 1)
                    length = mSize;
                return length;
            }
        }
        public void Clear()
        {
            PutInSub(mHeadIndexKey, -1L);
            PutInSub(mTailIndexKey, -1L);
        }
    }
}
